package com.dark.count;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

/**
 * Created by tengxue on 16-6-29.
 */
public class WordCountTopology {

    private static final String SENTENCE_SPOUT_ID="sentence-spout";
    private static final String SPLIT_BOLT_ID="split-bolt";
    private static final String COUNT_BOLT_ID="count-bolt";
    private static final String REPORT_BOLT_ID="report-bolt";
    private static final String TOPOLOGY_NAME="word-count-topology";

    public static void main(String[] args)throws Exception{
        SentenceSpout spout=new SentenceSpout();
        SplitSentenceBolt splitSentenceBolt=new SplitSentenceBolt();
        WordCountBolt wordCountBolt=new WordCountBolt();
        ReportBolt reportBolt=new ReportBolt();

        TopologyBuilder builder=new TopologyBuilder();
        builder.setSpout(SENTENCE_SPOUT_ID,spout);
        builder.setBolt(SPLIT_BOLT_ID,splitSentenceBolt).shuffleGrouping(SENTENCE_SPOUT_ID);
        //选择一个BOLT来执行统计的功能
        builder.setBolt(COUNT_BOLT_ID,wordCountBolt).fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
        builder.setBolt(REPORT_BOLT_ID,reportBolt).globalGrouping(COUNT_BOLT_ID);

        Config conf = new Config();
        //建议加上这行，使得每个bolt/spout的并发度都为1
        conf.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);

        LocalCluster cluster = new LocalCluster();

        //提交拓扑
        cluster.submitTopology(TOPOLOGY_NAME, conf, builder.createTopology());

        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        //结束拓扑
        cluster.killTopology(TOPOLOGY_NAME);

        cluster.shutdown();
    }
}
